/*
 * To change this template, choose Tools | Templates
 * and open the template in the editor.
 */
package sirtet.netplay;

import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStreamReader;
import java.io.PrintWriter;
import java.net.Socket;
import java.util.LinkedHashSet;
import java.util.LinkedList;
import sirtet.netplay.event.CommunicationDataEvent;
import sirtet.netplay.event.CommunicationEvent;
import sirtet.netplay.event.CommunicationListener;

/**
 * This is a wrapper of Socket and reader/writer that supports read and write
 * to the network peer asynchronously.
 * @author guangwei.zhu
 */
public class Communicator {

    private Socket socket;
    private String address;
    private int portnumber;
    private boolean activated = false;
    private BufferedReader reader;
    private PrintWriter writer;
    private LinkedHashSet<CommunicationListener> listeners = new LinkedHashSet<CommunicationListener>();
    private LinkedHashSet<CommunicationListener> toberemoved = new LinkedHashSet<CommunicationListener>();
    private boolean delayremoval = false;
    private MonitorThread monitor = new MonitorThread();
    private SenderThread sender = new SenderThread();

    protected class SenderThread extends Thread {

        private final Object listlock = new Object();
        private final Object sendlock = new Object();
        protected LinkedList<String> contents = new LinkedList<String>();

        public void send(String data) {
            synchronized (listlock) {
                // System.out.println("SEND: " + data);
                contents.addLast(data);
            }
            try {
                synchronized (sendlock) {
                    sendlock.notifyAll();
                }
            } catch (Exception ex) {
            }
        }

        @Override
        public void run() {
            Object[] buffered;
            boolean isCacheEmpty;

            while (true) {
                while (contents.isEmpty()) {
                    try {
                        synchronized (sendlock) {
                            sendlock.wait();
                        }
                    } catch (Exception ex) {
                    }
                }

                synchronized (listlock) {
                    buffered = contents.toArray();
                    contents.clear();
                }
                for (Object data : buffered) {
                    writer.println(data.toString());
                }
                if (!isConnected()) {
                    notifyDisconnected();
                    break;
                }
                writer.flush();
            }
        }
    }

    protected class MonitorThread extends Thread {

        private LinkedList<String> messageBuffer = new LinkedList<String>();

        @Override
        public void run() {
            String msg;

            while (true) {
                try {
                    msg = receive();    // Blocked until new data arrived
                    if (msg == null) {
                        // Return unexpected content
                        notifyDisconnected();
                        break;
                    }

                    while (!activated) {
                        yield();
                    }
                    if (!activated) {
                        // No listener, the data should be stored
                        messageBuffer.addLast(msg);
                    } else {
                        // Dump buffer first
                        while (!messageBuffer.isEmpty()) {
                            notifyDataReceived(messageBuffer.pollFirst());
                        }
                        notifyDataReceived(msg);
                    }
                } catch (IOException ex) {
                    // Disconnected, the thread cannot continue
                    notifyDisconnected();
                    break;
                }
            }
        }
    }

    public void activate() {
        this.activated = true;
    }

    private synchronized void delayedRemove() {
        if (toberemoved.size() > 0) {
            listeners.removeAll(toberemoved);
            toberemoved.clear();
        }
        delayremoval = false;
    }

    protected synchronized void notifyDataReceived(String data) {
        delayremoval = true;
        // System.out.println(data);
        for (CommunicationListener listener : listeners) {
            listener.communicationDataReceived(new CommunicationDataEvent(this, data));
        }
        delayedRemove();
    }

    protected synchronized void notifyDisconnected() {
        delayremoval = true;
        for (CommunicationListener listener : listeners) {
            listener.communicationDisconnected(new CommunicationEvent(this));
        }
        delayedRemove();
        disconnect();   // Enforce disconnection, detach all listeners
    }

    public void addCommunicationListener(CommunicationListener listener) {
        listeners.add(listener);
    }

    public synchronized void removeCommunicationListener(CommunicationListener listener) {
        if (delayremoval) {
            toberemoved.add(listener);
        } else {
            listeners.remove(listener);
        }
    }

    /**
     * Create a communicator using an existing Socket object.  The socket must
     * be already connected.
     * @param sock An existing, connected Socket object
     * @throws IOException if there are issues with the Socket object
     */
    public Communicator(Socket sock) throws IOException {
        socket = sock;

        address = sock.getInetAddress().getHostAddress();
        portnumber = sock.getPort();

        reader = new BufferedReader(new InputStreamReader(socket.getInputStream()));
        writer = new PrintWriter(socket.getOutputStream(), false);  // no autoflush

        monitor.start();
        sender.start();
    }

    /**
     * Create a communicator using the address and port number of the server.
     * A further call of connect() is necessary to activate the communicator.
     * @param addr Address of the server
     * @param port Port number of the server
     */
    public Communicator(String addr, int port) throws IOException {
        this(new Socket(addr, port));
    }

    /**
     * Disconnect underlying socket.  A disconnected communicator is useless
     * and a communicator should be constructed to reestablish connection.
     * @throws IOException if there are any networking problems.
     */
    public void disconnect() {
        try {
            if (!socket.isClosed()) {
                socket.close();
            }
        } catch (Exception ex) {
        } finally {
            listeners.clear();      // Detach all listeners
        }
    }

    /**
     * Detect if the underlying socket is connected.
     * @return true if the socket it connected.
     */
    public boolean isConnected() {
        return socket.isConnected();
    }

    /**
     * Send one line of string data to the peer.
     * @param data single line string data to be sent.
     * @throws IOException if the communicator is disconnected
     */
    public void send(String data) throws IOException {
        if (!this.isConnected()) {
            throw new IOException();
        }

        sender.send(data);
    }

    /**
     * Send one line of string data to the peer (does not throw IOException)
     * @param data single line string data to be sent
     */
    public void post(String data) {
        sender.send(data);
    }

    /**
     * Receive one line of string data from the peer.
     * @return a line of string data from the peer
     * @throws IOException
     */
    protected synchronized String receive() throws IOException {
        return reader.readLine();
    }
}
